package business

import (
	"encoding/json"
	"fcs-api/src/app/model"
	"os"
	"strconv"
	"sync"

	influx "github.com/influxdata/influxdb/client/v2"
)

var influxClient influx.Client

func InvokeCallback(invokeChan chan *model.Invoke) {
	influxConfig := influx.HTTPConfig{
		Addr:     model.Base.InfluxdbHost,
		Username: model.Base.InfluxdbUser,
		Password: model.Base.InfluxdbP6d,
	}
	var err error
	influxClient, err = influx.NewHTTPClient(influxConfig)
	if err != nil {
		logger.Error("failed to new influx client:%s", err.Error())
		os.Exit(1)
	}
	defer func() {
		_ = influxClient.Close()
	}()
	for {
		select {
		case invoke := <-invokeChan:
			invokeJson, _ := json.Marshal(invoke)
			logger.Info("InvokeCallback: get function callback, %v", string(invokeJson))
			saveInvokeData(invoke)
		}
	}
}

func saveInvokeData(invoke *model.Invoke) {
	// create point
	p, err := influx.NewPoint(
		model.InfluxInvokeTable,
		map[string]string{
			"creator":      invoke.Creator,
			"group_id":     invoke.GroupId,
			"group_name":   invoke.GroupName,
			"func_id":      invoke.FuncId,
			"func_name":    invoke.FuncName,
			"version_name": invoke.VersionName,
			"invoke_mode":  invoke.InvokeMode,
			"invoke_type":  invoke.InvokeType,
			"return_code":  strconv.Itoa(invoke.ReturnCode),
		},
		map[string]interface{}{
			"id":               invoke.Id,
			"runtime_duration": invoke.RuntimeDuration,
			"invoke_duration":  invoke.InvokeDuration,
			"invoke_source":    invoke.InvokeSource,
			"memory_usage":     invoke.MemoryUsage,
			"memory_limit":     invoke.MemoryLimit,
			"compute_power":    float64(invoke.MemoryLimit) / 1024 / 1024 * float64(invoke.RuntimeDuration) / 1000,
			"return_msg":       invoke.ReturnMsg,
			"log":              invoke.Log,
		},
		invoke.StartTime)
	if err != nil {
		logger.Error("saveInvokeData: failed to create influx points,%s", err.Error())
		return
	}
	// write asynchronously
	bps, err := influx.NewBatchPoints(influx.BatchPointsConfig{
		Database:        model.InfluxDBName,
		RetentionPolicy: model.InfluxRetentionPolicy,
	})
	if err != nil {
		logger.Error("saveInvokeData: failed to new influx batch points:%s", err.Error())
		return
	}
	bps.AddPoint(p)
	err = influxClient.Write(bps)
	if err != nil {
		logger.Error("saveInvokeData: failed write points:%s", err.Error())
		return
	}
}

func GetLogList(query *model.LogQuery) (*model.LogResult, error) {
	logResult := new(model.LogResult)
	invokeList, err := model.GetInvokeList(query)
	if err != nil {
		_ = logger.Error("GetLogList: get invokeList from db error, %v", err)
		return new(model.LogResult), err
	}
	invokeCount, err := model.GetInvokeListCount(query)
	if err != nil {
		_ = logger.Error("GetLogList: get invokeCount from db error, %v", err)
		return new(model.LogResult), err
	}
	logResult.Result = invokeList
	//logInvokeList, err := getEsLog(invokeList)
	//if err != nil {
	//	_ = logger.Error("GetLogList: get log from es error, %v", err)
	//	logResult.Result = invokeList
	//} else {
	//	logResult.Result = logInvokeList
	//}
	logResult.RecordsFiltered = len(invokeList)
	logResult.TotalCount = int(invokeCount)

	return logResult, nil
}

func getEsLog(invokeList []*model.Invoke) ([]*model.Invoke, error) {

	num := len(invokeList)
	if num == 0 {
		return invokeList, nil
	}
	//esClient, err := es.GetEsClient()
	//if err != nil {
	//	_ = logger.Error("getEsLog: get es client error, %v", err)
	//	return nil, err
	//}

	var wg sync.WaitGroup
	wg.Add(num)
	for _, invoke := range invokeList {
		invoke := invoke
		go func() {
			defer wg.Done()
			//log, err := QueryEs(esClient, strings.Replace(invoke.Id, "-", "", -1)) // 去掉requestid中的横线
			//if err != nil {
			//	_ = logger.Error("getEsLog: get log from es error, %v", err)
			//	return
			//}
			invoke.Log = invoke.ReturnMsg
		}()
	}
	wg.Wait()

	return invokeList, nil
}
